package com.hudson.broker.handler.process;

import com.hudson.broker.handler.Process;
import com.hudson.broker.monitor.ThreadPoolHelper;
import com.hudson.broker.utils.StoreUtil;
import com.hudson.codec.domain.MessageExt;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.util.CollectionUtil;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;


/**
 * 目前先用时间轮处理重试消息
 */
public class ConsumerRetryProcess implements Process {


    private static final Logger log = LoggerFactory.getLogger(ConsumerRetryProcess.class);

    private static final int MAX_RETRY_NUM = 10;

    private static final Map<Integer, Long> RETRY_TIME = Map.of(1, 5L,
            2, 10L,
            3, 30L,
            4, 60L,
            5, 120L,
            6, 300L,
            7, 600L,
            8, 900L,
            9, 1800L,
            10, 3600L);


    @Override
    public RemotingCommand process(ChannelHandlerContext context, RemotingCommand remotingCommand) {
        if (CollectionUtil.isEmpty(remotingCommand.getBody())) {
            return null;
        }
        List<MessageExt> messageExtList = MessageExt.parse2List(remotingCommand.getBody());
        if (CollectionUtil.isEmpty(messageExtList)) {
            return null;
        }

        for (MessageExt messageExt : messageExtList) {
            if (messageExt.getReconsumeTimes() > MAX_RETRY_NUM) {
                StoreUtil.putDLQMessage(messageExt);
                return null;
            }
            ThreadPoolHelper.WHEEL_TIMER.newTimeout((a) -> {
                StoreUtil.putRetryMessage(messageExt);
            }, RETRY_TIME.get(messageExt.getReconsumeTimes()), TimeUnit.SECONDS);
        }
        return RemotingCommand.builder()
                .type(MessageTypeEnum.SUCCESS_MESSAGE.getType())
                .opaque(remotingCommand.getOpaque())
                .build();
    }
}
